package com.tmt.im.service.interfaces.impl;

import com.tmt.im.common.pojo.*;
import com.tmt.im.common.store.SocketMessage;
import com.tmt.im.common.store.StoreSqlHelper;
import com.tmt.im.service.interfaces.IDataSender;
import com.tmt.im.service.websocket.SessionContextManager;
import jakarta.websocket.Session;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.cache.query.FieldsQueryCursor;
import org.apache.ignite.cache.query.SqlFieldsQuery;
import org.springframework.util.CollectionUtils;

import java.io.Serial;
import java.io.Serializable;
import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * @Description TODO
 * @Author wuyi
 * @Date 2024/8/17 9:21
 * @Version 1.0
 **/
@Slf4j
@Setter
public class ChatDataSender implements IDataSender, Serializable {

    @Serial
    private static final long serialVersionUID = 7279436487085622238L;

    private IgniteCache<SessionMemberKey, ChatSessionMemberPoJo> sessionMemberCache;

    private IgniteCache<Long, MemberStatusPoJo> memberStatusCache;

    private IgniteCache<ChatDataKey, ChatDataPoJo> dataCache;

    private SessionContextManager contextManager;

    private int pushDataCount;

    public void init(IgniteCache<SessionMemberKey, ChatSessionMemberPoJo> sessionMemberCache,
                     IgniteCache<Long, MemberStatusPoJo> memberStatusCache,
                     IgniteCache<ChatDataKey, ChatDataPoJo> dataCache,
                     SessionContextManager contextManager,
                     int pushDataCount) {
        this.sessionMemberCache = sessionMemberCache;
        this.memberStatusCache = memberStatusCache;
        this.dataCache = dataCache;
        this.contextManager = contextManager;
        this.pushDataCount = pushDataCount;
    }

    @Override
    public void sendData(ChatSessionMemberPoJo sessionMemberPoJo) {
        try {
            Long sessionId = sessionMemberPoJo.getSessionId();
            Long memberId = sessionMemberPoJo.getMemberId();

            Session session = contextManager.getSessionByUserId(memberId);
            if (session == null) {
                return;
            }

            SocketMessage msg = new SocketMessage();
            msg.setSessionId(sessionId);
            msg.setNewDataCount(0);

            MemberStatusPoJo statusPoJo = memberStatusCache.get(memberId);
            if (statusPoJo != null && Objects.equals(statusPoJo.getSessionId(), sessionId)) {
                //读取最新的最多20条数据, 不论是否已经删除了都要推送。
                String tmpSql = MessageFormat.format("select * from {0} where mid > ? and sessionId = ? order by mid desc limit ?",
                        ChatDataPoJo.CACHE_NAME);
                SqlFieldsQuery tmpQuery = new SqlFieldsQuery(tmpSql).setArgs(sessionMemberPoJo.getLastDataId(), sessionId, pushDataCount);
                List<ChatDataPoJo> ChatDataPoJos = new ArrayList<>();
                try (FieldsQueryCursor<List<?>> cursor = dataCache.query(tmpQuery)) {
                    StoreSqlHelper.convertTo(cursor, ChatDataPoJos, ChatDataPoJo.class);
                }
                if (!CollectionUtils.isEmpty(ChatDataPoJos)) {
                    //更新最后推送的数据记录ID。
                    sessionMemberPoJo.setInsert(false);
                    sessionMemberCache.replace(sessionMemberPoJo.getKey(), sessionMemberPoJo);

                    //排除x.getLastDataId()记录，并升序排序。
                    List<ChatDataPoJo> validList = ChatDataPoJos.stream()
                            .sorted(Comparator.comparingLong(ChatDataPoJo::getId))
                            .collect(Collectors.toList());
                    msg.setDataList(validList);
                    //判断是否还有数据未推送过，如果有则向客户端推送未推送数据的数量。
                    if (validList.size() == pushDataCount) {
                        long newDataCount = totalNewDataCount(sessionId, sessionMemberPoJo.getLastDataId(), validList.get(0).getId());
                        msg.setNewDataCount(newDataCount);
                    }
                }
            } else {
                //统计未推送的数据数量。
                long newDataCount = totalNewDataCount(sessionId, sessionMemberPoJo.getLastDataId(), null);
                msg.setNewDataCount(newDataCount);
            }

            //推送数据。
            try {
                session.getAsyncRemote().sendObject(msg);
            } catch (Exception ex) {
                //异常后直接关闭该连接。
                session.close();
            }
        } catch (Exception ex) {
            log.error("向群组中用户分发聊天数据出现异常", ex);
        }
    }

    @Override
    public void sendData(Long sessionId, Long memberId) {
        String tmpSql = MessageFormat.format("select * from {0} where sessionId = ? and memberId = ? and agreeStatus = ?",
                ChatSessionMemberPoJo.CACHE_NAME);
        SqlFieldsQuery tmpQuery = new SqlFieldsQuery(tmpSql).setArgs(sessionId, memberId, 1);
        List<ChatSessionMemberPoJo> sessionMemberPoJos = new ArrayList<>();
        try (FieldsQueryCursor<List<?>> tmpCursor = sessionMemberCache.query(tmpQuery)) {
            StoreSqlHelper.convertTo(tmpCursor, sessionMemberPoJos, ChatSessionMemberPoJo.class);
        }
        if (!CollectionUtils.isEmpty(sessionMemberPoJos)) {
            sendData(sessionMemberPoJos.get(0));
        }
    }

    private long totalNewDataCount(Long sessionId, Long startId, Long endId) {
        long newDataCount = 0;
        String tmpSql = MessageFormat.format("select COUNT(*) from {0} where mid > ?{1}and sessionId = ?",
                ChatDataPoJo.CACHE_NAME, endId != null ? " and mid < ? " : " ");
        SqlFieldsQuery tmpQuery = endId != null ? new SqlFieldsQuery(tmpSql).setArgs(startId, endId, sessionId) :
                new SqlFieldsQuery(tmpSql).setArgs(startId, sessionId, false);
        try (FieldsQueryCursor<List<?>> tmpCursor = dataCache.query(tmpQuery)) {
            List<List<?>> tmpList = tmpCursor.getAll();
            newDataCount = (long) tmpList.get(0).get(0);
        }
        return newDataCount;
    }

}
